package com.only4play.portal.cp.message;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.only4play.portal.common.error.PortalErrorExceptionHandler;
import com.only4play.portal.common.api.MessageDuplicateChecker;
import com.only4play.portal.common.api.MessageInMemoryDuplicateChecker;
import com.only4play.portal.common.session.PortalSessionManager;
import com.only4play.portal.common.util.LogExceptionHandler;
import com.only4play.portal.cp.api.CpService;
import com.only4play.portal.cp.bean.message.AbstractCpMessage;
import com.only4play.portal.cp.bean.message.AbstractCpOutMessage;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 *
 * <pre>
 *
 * 通过代码化的配置，把消息交给handler处理
 *
 * 说明：
 * 1. 配置路由规则时要按照从细到粗的原则，否则可能消息可能会被提前处理
 * 2. 默认情况下消息只会被处理一次，除非使用CpMessageRouter.next()
 * 3. 规则结束需使用CpMessageRouter.next()或者CpMessageRouter.end()
 *
 * </pre>
 *
 * @author liyuncong
 * @version 1.0
 * @file CpMessageRouter
 * @brief CpMessageRouter
 * @details 消息处理路由
 * @date 2024-06-07
 * <p>
 *
 * Edit History
 * ----------------------------------------------------------------------------
 * DATE                     NAME               DESCRIPTION
 * 2024-06-07               liyuncong          Created
 */

@Getter
@Setter
@Slf4j
public abstract class CpMessageRouter<CpOutMessage extends AbstractCpOutMessage, CpMessage extends AbstractCpMessage> {

    protected static final int DEFAULT_THREAD_POOL_SIZE = 100;
    protected final List<CpMessageRouterRule<CpOutMessage, CpMessage>> rules = new ArrayList<>();

    protected final CpService portalCpService;

    protected ExecutorService executorService;

    protected MessageDuplicateChecker messageDuplicateChecker;

    protected PortalSessionManager sessionManager;

    protected PortalErrorExceptionHandler exceptionHandler;

    public CpMessageRouter(CpService cpService) {
        this.portalCpService = cpService;
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("CpMessageRouter-pool-%d").build();
        this.executorService = new ThreadPoolExecutor(DEFAULT_THREAD_POOL_SIZE, DEFAULT_THREAD_POOL_SIZE,
            0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), namedThreadFactory);
        this.messageDuplicateChecker = new MessageInMemoryDuplicateChecker();
        this.sessionManager = cpService.getSessionManager();
        this.exceptionHandler = new LogExceptionHandler();
    }

    /**
     * 开始一个新的Route规则
     *
     * @return
     */
    public abstract CpMessageRouterRule<CpOutMessage, CpMessage> rule();

    public CpOutMessage route(final CpMessage cpMessage) {
        return this.route(cpMessage, new HashMap<>(2));
    }

    public CpOutMessage route(final CpMessage cpMessage, final Map<String, Object> context) {
        if (isMsgDuplicated(cpMessage)) {
            // 如果是重复消息，那么就不做处理
            return null;
        }

        final List<CpMessageRouterRule<CpOutMessage, CpMessage>> matchRules = new ArrayList<>();
        // 收集匹配的规则
        for (final CpMessageRouterRule<CpOutMessage, CpMessage> rule : this.rules) {
            if (rule.attemptMatch(cpMessage)) {
                matchRules.add(rule);
                if (!rule.isReEnter()) {
                    break;
                }
            }
        }

        if (matchRules.isEmpty()) {
            return null;
        }

        CpOutMessage outMessage = null;
        final List<Future> futures = new ArrayList<>();
        for (final CpMessageRouterRule<CpOutMessage, CpMessage> rule : matchRules) {
            // 返回最后一个非异步的rule的执行结果
            if (rule.isAsync()) {
                futures.add(
                    this.executorService.submit(() -> {
                        rule.service(cpMessage, context, CpMessageRouter.this.portalCpService, CpMessageRouter.this.sessionManager, CpMessageRouter.this.exceptionHandler);
                    })
                );
            } else {
                outMessage = rule.service(cpMessage, context, this.portalCpService, this.sessionManager, this.exceptionHandler);
                // 在同步操作结束，session访问结束
                log.debug("End session access: async=false");
                sessionEndAccess(cpMessage);
            }
        }

        if (!futures.isEmpty()) {
            this.executorService.submit(() -> {
                for (Future future : futures) {
                    try {
                        future.get();
                        log.debug("End session access: async=true");
                        // 异步操作结束，session访问结束
                        sessionEndAccess(cpMessage);
                    } catch (InterruptedException exception) {
                        log.error("Error happened when wait task finish", exception);
                        Thread.currentThread().interrupt();
                    } catch (ExecutionException exception) {
                        log.error("Error happened when wait task finish", exception);
                    }
                }
            });
        }

        return outMessage;
    }

    /**
     * 对session的访问结束
     */
    protected abstract void sessionEndAccess(CpMessage cpMessage);

    protected abstract boolean isMsgDuplicated(CpMessage cpMessage);

}
